package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.common.kafkaTopics;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author Lec
 * @date 2022/9/5 14:23
 */

public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {
        //TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(905L));

        //TODO 2. 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */
        //TODO 3. 从kafka的topic_db读取数据
        String topicName = "topic_db";
        String groupID = "dwd_trade_order_pre_process";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` bigint,\n" +
                "  `xid` string,\n" +
                "  `commit` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string>,\n" +
                "   pt AS PROCTIME() \n" +
                ")"
                + KafkaUtil.getKafkaDDL(topicName, groupID));

        //TODO 4. 过滤出order_detail数据
        Table orderDetailTable = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['order_id'] order_id,\n" +
                "  `data`['course_id'] course_id,\n" +
                "  `data`['course_name'] course_name,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `pt`,\n" +
                "  ts\n" +
                "from topic_db\n" +
                "where `table`='order_detail'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("order_detail",orderDetailTable);

        //TODO 5. 过滤出order_info数据
        Table orderInfoTable = tableEnv.sqlQuery("select \n" +
                "  `data`['id'] id,\n" +
                "  `data`['order_status'] order_status,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['session_id'] session_id,\n" +
                "  `data`['province_id'] province_id,\n" +
                "   `type`,\n" +
                "   `old`,\n" +
                "  `ts`,\n" +
                "  `pt`\n" +
                "from topic_db\n" +
                "where`table`='order_info'\n" +
                "and (`type`='insert' or `type`='update')");
        tableEnv.createTemporaryView("order_info",orderInfoTable);

        //读取page_log数据
        tableEnv.executeSql("create table page_log \n" +
                "    (\n" +
                "  `common` map<string,string>,\n" +
                "  `page` map<string,string>,\n" +
                "  `ts` bigint\n" +
                "    )"+ KafkaUtil.getKafkaDDL(kafkaTopics.DWD_TRAFFIC_PAGE_LOG,groupID));
        //过滤出所需要的信息
        Table pageLogTable = tableEnv.sqlQuery("select\n" +
                "    `common`['sc'] sc,\n" +
                "    `common`['uid'] uid,\n" +
                "    `common`['sid'] sid\n" +
                "from page_log\n" +
                "where `page`['page_id'] = 'order'");
        tableEnv.createTemporaryView("pageLog",pageLogTable);

// TODO 9 join读取的3张表格
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "  od.id,\n" +
                "  od.order_id,\n" +
                "  od.course_id,\n" +
                "  od.course_name,\n" +
                "  od.user_id,\n" +
                "  od.origin_amount,\n" +
                "  od.coupon_reduce,\n" +
                "  od.final_amount,\n" +
                "  od.create_time,\n" +
                "  od.update_time,\n" +
                "  od.ts od_ts,\n" +
                "  oi.order_status,\n" +
                "  oi.session_id,\n" +
                "  pl.sc,\n" +
                "  oi.province_id,\n" +
                "  oi.ts oi_ts,\n" +
                "  oi.pt,\n" +
                "   oi.`type`,\n" +
                "   oi.`old`,\n" +
                "  current_row_timestamp() row_op_ts\n" +
                "from order_detail od\n" +
                "join order_info oi\n" +
                "on od.order_id=oi.id\n" +
                "join pageLog pl\n" +
                "on oi.session_id = pl.sid");
        tableEnv.createTemporaryView("result_table",resultTable);

        // TODO 10 写入到kafka新的主题中dwd_trade_order_pre_process
        tableEnv.executeSql("create table kafka_sink(\n" +
                "  id string,\n" +
                "  order_id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  user_id string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  final_amount string,\n" +
                "  create_time string,\n" +
                "  update_time string,\n" +
                "  od_ts  bigint,\n" +
                "  order_status string,\n" +
                "  session_id string,\n" +
                "  sc string,\n" +
                "  province_id string,\n" +
                "  oi_ts bigint,\n" +
                "  pt TIMESTAMP_LTZ(3),\n" +
                "   `type` string,\n" +
                "   `old` map<string,string>,\n" +
                "  row_op_ts TIMESTAMP_LTZ(3)," +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ")" + KafkaUtil.getUpsertKafkaSinkDDL("dwd_trade_order_pre_process"));

        tableEnv.executeSql("insert into kafka_sink select * from result_table");


        /*
{
    "id": "48321",
    "order_id": "44457",
    "course_id": "42",
    "course_name": "Maven",
    "user_id": "28",
    "origin_amount": "200.0",
    "coupon_reduce": "0.0",
    "final_amount": "200.0",
    "create_time": "2022-09-05 15:00:54",
    "update_time": null,
    "od_ts": 1592118054,
    "order_status": "1002",
    "session_id": "e337359c-5ab6-480b-a356-f23e294dd2e4",
    "province_id": "4",
    "oi_ts": 1592118054,
    "pt": "2022-09-05 07:00:55.236Z",
    "type": "update",
    "old": {
        "update_time": null,
        "order_status": "1001"
    },
    "row_op_ts": "2022-09-05 07:00:55.236Z"
}
         */
        // TODO 执行任务
//        env.execute();
    }
}
